package cn.csg.ams.collect2.poptopics.datahub;


import com.aliyun.datahub.client.exception.AuthorizationFailureException;
import com.aliyun.datahub.client.exception.DatahubClientException;
import com.aliyun.datahub.client.exception.InvalidParameterException;
import com.aliyun.datahub.client.exception.NoPermissionException;
import com.aliyun.datahub.client.exception.SubscriptionOfflineException;
import com.aliyun.datahub.client.exception.SubscriptionOffsetResetException;
import com.aliyun.datahub.client.exception.SubscriptionSessionInvalidException;
import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.client.model.TupleRecordData;
import com.aliyun.datahub.clientlibrary.config.ConsumerConfig;
import com.aliyun.datahub.clientlibrary.consumer.Consumer;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Data
public abstract class BaseDatahubConsumer extends Thread {

    private static Logger logger = LoggerFactory.getLogger(BaseDatahubConsumer.class);

    private String endpoint;

    private String accessId;

    private String accessKey;

    private String projectName;

    private String topicName;

    private String subId;

    @Override
    public void run() {
        consume();
    }

    /**
     * 协同消费
     */
    private void consume() {
        ConsumerConfig config = new ConsumerConfig(endpoint, accessId, accessKey);
        Consumer consumer = new Consumer(projectName, topicName, subId, config);

        int maxRetry = 3;
        try {
            while (true) {
                try {
                    RecordEntry record = consumer.read(maxRetry);
                    // 处理数据
                    if (record != null) {
                        TupleRecordData data = (TupleRecordData) record.getRecordData();
                        dataHandle(data, record.getSystemTime());

                        record.getKey().ack();
                    } else {
                        System.out.println(Thread.currentThread().getName() + ": " + topicName + " read null");
                    }
                } catch (SubscriptionOffsetResetException e) {
                    // 点位被重置， 重新初始化consumer
                    try {
                        consumer.close();
                        consumer = new Consumer(projectName, topicName, subId, config);
                    } catch (DatahubClientException e1) {
                        logger.error("初始化失败", e);
                    }
                } catch (InvalidParameterException |
                         SubscriptionOfflineException |
                         SubscriptionSessionInvalidException |
                         AuthorizationFailureException |
                         NoPermissionException e) {
                    logger.error("读取数据失败", e);
                    throw e;
                } catch (DatahubClientException e) {
                    logger.error("读取数据失败， 重试", e);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    }
                } catch (Exception e) {
                    logger.error("读取数据失败， 重试", e);
                }
                // catch (BadSqlGrammarException | DataIntegrityViolationException e) {
                //     logger.error("sql执行失败", e);
                // }
            }
        } catch (Throwable e) {
            logger.error("读取失败", e);
        } finally {
            // 确保资源正确释放
            // 会提交已ack的点位
            consumer.close();
        }
    }


    /**
     * 处理数据
     */
    abstract void dataHandle(TupleRecordData data, long systemTime);

}
